package com.zx.learn.flink.cdc;

//import cn.it.sz22.utils.ModelUtil;
//import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
//import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
//import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
//import org.apache.flink.api.common.functions.MapFunction;
//import org.apache.flink.streaming.api.datastream.DataStreamSource;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;

/**
 * https://blog.csdn.net/yang_shibiao/article/details/122774389
 *从业务库拉取binlog数据
 */
public class GmallCDC {

//    public static Logger logger = LoggerFactory.getLogger(GmallCDC.class);
//    public static final String ODS_BASE_DB_TOPIC_NAME = ModelUtil.getConfigValue("kafka.topic.ods.base.db");
//
//    public static void main(String[] args) throws Exception {
//
//        String applicationName = "gmall-cdc";
//        long interval = 5000L;
//
//        // 1.获取执行环境，并配置checkpoint
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        FlinkUtil.deployRocksdbCheckpoint(env, applicationName, interval, true);
//
//
//        // 2.通过FlinkCDC构建SourceFunction并读取数据
//        DebeziumSourceFunction<String> sourceFunction = MySQLSource.<String>builder()
//                .hostname(ModelUtil.getConfigValue("mysql.hostname"))
//                .port(Integer.parseInt(ModelUtil.getConfigValue("mysql.port")))
//                .username(ModelUtil.getConfigValue("mysql.username"))
//                .password(ModelUtil.getConfigValue("mysql.password"))
//                .databaseList(ModelUtil.getConfigValue("mysql.database.gmall"))
//                .deserializer(new CustomerDeserialization())
//                .startupOptions(StartupOptions.initial())
//                .build();
//        DataStreamSource<String> streamSource = env.addSource(sourceFunction);
//
//
//        //3.对数据进行日志打印，并将数据输出到Kafka中
//        streamSource
//                .map(new MapFunction<String, String>() {
//                    @Override
//                    public String map(String value) throws Exception {
//                        logger.warn(value);
//                        return value;
//                    }
//                })
//                .addSink(MyKafkaUtil.getKafkaProducerExactlyOnce(ODS_BASE_DB_TOPIC_NAME));
//
//
//        //4.启动任务
//        env.execute(applicationName);
//
//
//    }

}
